import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * 任务一：消费行为统计
 * 统计每个商家的优惠券使用情况（线上线下分开）
 * 输出格式：<Merchant_id> TAB <负样本数> TAB <普通消费数> TAB <正样本数>
 */
public class Task1MerchantCouponStats {

    // 线下数据Mapper
    public static class OfflineMapper extends Mapper<Object, Text, Text, Text> {
        private Text merchantKey = new Text();
        private Text value = new Text();

        public void map(Object key, Text line, Context context) throws IOException, InterruptedException {
            String[] fields = line.toString().split(",", -1);
            if (fields.length < 7 || fields[0].equals("User_id")) {
                return; // 跳过表头和无效数据
            }

            String merchantId = fields[1];    // Merchant_id
            String couponId = fields[2];      // Coupon_id
            String date = fields[6];          // Date

            if (merchantId == null || merchantId.equals("null") || merchantId.isEmpty()) {
                return;
            }

            // 线下标记：offline
            merchantKey.set("offline_" + merchantId);

            boolean hasCoupon = !(couponId.equals("null") || couponId.isEmpty());
            boolean hasDate = !(date.equals("null") || date.isEmpty());

            if (!hasCoupon && hasDate) {
                // 未领取优惠券直接消费（普通消费）
                value.set("normal:1");
                context.write(merchantKey, value);
            } else if (hasCoupon) {
                // 只要有优惠券ID
                if (hasDate) {
                    // 领取并使用（正样本）
                    value.set("positive:1");
                    context.write(merchantKey, value);
                } else {
                    // 领取未使用（负样本）
                    value.set("negative:1");
                    context.write(merchantKey, value);
                }
            }
        }
    }

    // 线上数据Mapper
    public static class OnlineMapper extends Mapper<Object, Text, Text, Text> {
        private Text merchantKey = new Text();
        private Text value = new Text();

        public void map(Object key, Text line, Context context) throws IOException, InterruptedException {
            String[] fields = line.toString().split(",", -1);
            if (fields.length < 7 || fields[0].equals("User_id")) {
                return;
            }

            // 线上数据有Action字段,字段索引不同
            String merchantId = fields[1];    // Merchant_id
            // fields[2] 是 Action（线上独有）
            String couponId = fields[3];      // Coupon_id
            String date = fields[6];          // Date

            if (merchantId == null || merchantId.equals("null") || merchantId.isEmpty()) {
                return;
            }

            // 线上标记：online
            merchantKey.set("online_" + merchantId);

            boolean hasCoupon = !(couponId.equals("null") || couponId.isEmpty());
            boolean hasDate = !(date.equals("null") || date.isEmpty());

            if (!hasCoupon && hasDate) {
                // 未领取优惠券直接消费（普通消费）
                value.set("normal:1");
                context.write(merchantKey, value);
            } else if (hasCoupon) {
                // 只要有优惠券ID
                if (hasDate) {
                    // 领取并使用（正样本）
                    value.set("positive:1");
                    context.write(merchantKey, value);
                } else {
                    // 领取未使用（负样本）
                    value.set("negative:1");
                    context.write(merchantKey, value);
                }
            }
        }
    }

    // Reducer
    public static class StatsReducer extends Reducer<Text, Text, Text, Text> {
        private Text result = new Text();

        public void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            int negative = 0;  // 负样本数
            int normal = 0;    // 普通消费数
            int positive = 0;  // 正样本数

            for (Text val : values) {
                String[] parts = val.toString().split(":");
                if (parts.length == 2) {
                    int count = Integer.parseInt(parts[1]);
                    if (parts[0].equals("negative")) {
                        negative += count;
                    } else if (parts[0].equals("normal")) {
                        normal += count;
                    } else if (parts[0].equals("positive")) {
                        positive += count;
                    }
                }
            }

            result.set(negative + "\t" + normal + "\t" + positive);
            context.write(key, result);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Task1 Merchant Coupon Stats");
        job.setJarByClass(Task1MerchantCouponStats.class);

        // 设置Reducer
        job.setReducerClass(StatsReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        // 多输入：线下数据
        MultipleInputs.addInputPath(job, new Path(args[0]),
                TextInputFormat.class, OfflineMapper.class);

        // 多输入：线上数据
        MultipleInputs.addInputPath(job, new Path(args[1]),
                TextInputFormat.class, OnlineMapper.class);

        FileOutputFormat.setOutputPath(job, new Path(args[2]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
